package com.weiyi.biz.service;

import com.alibaba.fastjson.JSONObject;
import com.weiyi.biz.domain.HelmetLog;
import com.ys.product.ysmq.front.msg.AbstractConsumerMessage.ConsumerCallBack;
import com.ys.product.ysmq.front.msg.StandardConsumerMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * <p>  安全帽轨迹 简单版
 * 异步处理消息demo. 消息管道的回调接口:ConsumerCallBack,是和消息消费在一个线程中的,所以用户拿到消息后,要放到本地队列用单独的线程或线程池(消息量非常大的时候)去处理.
 * 这个demo演示如何异步处理消息,抽象的说明怎么异步处理,用户可借鉴这个,结合自己的场景,实现适合自己的异步处理逻辑.
 * </p>
 *
 * @author jiangyuechao 2018年8月13日 上午11:09:48
 * @version V1.0
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify by user: {修改人} 2018年8月13日
 * @modify by reason:{方法名}:{原因}
 */
@Component
@Order(98)
public class AsyncConsumerMessage implements ApplicationRunner {

    @Autowired
    IHelmetLogService helmetLogService;

    private static final Logger log = LoggerFactory.getLogger(AsyncConsumerMessage.class);

    private static String appKey = "c6f6470a76644b8893b5160cf357e454";
    private static String appSecret = "8f9cc31607ac1da176bad1d6a7051a9c";
    private static String group = "group3";

    // 异步处理消息的本地队列.
    private static BlockingQueue<List<Object>> msgQueue = new LinkedBlockingQueue<List<Object>>();


    @Override
    public void run(ApplicationArguments args) throws Exception {

        log.info("安全帽开始监听---------------------");

        // 启动异步线程
        AsyncHandleThread asyncHandleThread = new AsyncHandleThread();
        asyncHandleThread.start();
        // 启动SDK消费消息
        init();
    }

    public static void init() {
        // 开放平台的url,这是test环境的url https://test.ys7.com:65
        String path = "https://open.ys7.com";
        //
        // 设置你的appkey,appSecret,group
        StandardConsumerMessage consumerMessage = new StandardConsumerMessage(path, appKey, appSecret, group);
        // 设置调用接口的间隔时间,ms
        consumerMessage.setConsumeIntervalTime(10000);
        // 设置消费消息的回调接口,建议把消息放入本地队列,不要在回调函数里做业务处理.
        consumerMessage.setConsumerCallBack(new ConsumerCallBack() {
            @Override
            public void consumerCall(List<Object> msg) {
                try {
                    // 使用队列异步处理消息
                    msgQueue.put(msg);
                } catch (InterruptedException e) {
                    // 用户这里需要处理异常
                    log.error(String.format("放入队列失败,msgs:", msg), e);
                }
            }
        });
        consumerMessage.initClient();
    }


    /**
     * <p>
     * 异步处理消息的线程,消息量非常大的时候,请使用线程池.
     * </p>
     */
     class AsyncHandleThread extends Thread {
        AtomicBoolean isRunning = new AtomicBoolean(true);

        public AsyncHandleThread() {
            super("Async-HandleMsg-Thread");
        }

        @Override
        public void run() {
            while (isRunning.get()) {
                try {
                    List<Object> msgs = msgQueue.take();
                    if (msgs.size() > 0) {
                        log.info("异步处理消息,消息数量:{}", msgs.size());
                        userProcessMsgFunction(msgs);
                    }
                } catch (Exception e) {
                    // 用户这里需要处理异常
                    log.error("线程中断异常", e);
                }
            }
        }

        // 关闭线程
        public void shutdown() {
            isRunning.set(false);
        }
    }

    /**
     * <p>
     * 用户自己的处理消息的函数
     * </p>
     *
     * @param msgs
     * @author jiangyuechao 2018年8月13日 上午11:21:25
     */
    SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

     void userProcessMsgFunction(List<Object> msgs) throws ParseException {
        for (Object object : msgs) {
            log.info("my readMessageHandle object:{}", object);

            JSONObject obj = (JSONObject) object;

            //获取消息时间
            JSONObject hObj=JSONObject.parseObject(obj.getString("header"));
            double time2 = Double.valueOf(hObj.getString("messageTime"));
            String date = sdf.format(time2);
            Date  dd= sdf.parse(date);


            //获取经纬度
            String body = obj.getString("body").replaceAll("\\\\t", "").replaceAll("\\\\n", "").replaceAll("\\\\", "");
            body=body.substring(12,body.length()-2);

            System.out.println(body);

            JSONObject playload=JSONObject.parseObject(body);

            String helmetInfo=playload.getString("FourGSafetyHelmetInfo");
            helmetInfo=helmetInfo.substring(13,helmetInfo.length()-1);
            System.out.println(helmetInfo);

            if (helmetInfo.contains("GPS")){
                JSONObject gps=JSONObject.parseObject(helmetInfo);
                JSONObject add=JSONObject.parseObject(gps.getString("GPS"));

                String p1=add.getString("latitude");
                String p2=add.getString("longitude");

                System.out.println(p2+","+p1);


                if (!"0".equals(p1) && !"0".equals(p2)) {
                    HelmetLog helmetLog = new HelmetLog();
                    helmetLog.setLatitude(p1);
                    helmetLog.setLongitude(p2);
                    helmetLog.setOperateTime(dd);
                    helmetLogService.insertHelmetLog(helmetLog);
                }

            }

        }
    }
}
